package cn.doitedu.dw_etl.utils

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.{DataFrame, Dataset, Encoders, Row, SparkSession}

class Person(val id:Int,val name:String){

}

object AreaDictGeo {
  def main(args: Array[String]): Unit = {

    if(args.length<1){
      println(
        """
          |gps坐标转geohash，地理位置字典转换程序
          |需要传入如下参数：
          |参数1：结果保存路径
          |
          |""".stripMargin)

      sys.exit(1)  // 退出码  在linux中可以通过如下方式获取上一个命令的退出码： [linux shell] echo $?
    }

    val spark = SparkSession.builder().appName("地理位置字典geohash编码转化").master("local").getOrCreate()

    // 读取mysql中的原始地理位置字典表
    val props = new Properties()
    props.load(AreaDictGeo.getClass.getClassLoader.getResourceAsStream("prod/db.properties"))

    val df: DataFrame = spark.read.jdbc("jdbc:mysql://localhost:3306/realtimedw?characterEncoding=utf8&useUnicode=true", "gps_area", props)

    import spark.implicits._
    val res = df.map(row=>{
      val lat = row.getAs[Double]("lat")
      val lng = row.getAs[Double]("lng")
      val province = row.getAs[String]("province")
      val city = row.getAs[String]("city")
      val region = row.getAs[String]("region")

      // 把经纬度坐标，转换成geohash编码
      val geo = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 5)

      (geo,province,city,region)
    }).toDF("geo","province","city","region")

    res.write.parquet(args(0))

    spark.close()

  }

}
